package com.tcs.monitoring.servers.impl;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.SuccessCallback;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.tcs.kafka.producer.sender.IKafkaSender;
import com.tcs.model.email.MailSenderInfo;
import com.tcs.model.monitoring.info.AbstractMonitoringInfo;
import com.tcs.model.rule.RuleBean;
import com.tcs.monitoring.servers.ISendService;
import com.tcs.rpc.service.IEmailService;
import com.tcs.server.rule.IRuleService;
import com.tcs.util.constant.SystemConstants;

public class SendServiceImpl implements ISendService {
	
	private final Logger logger = LoggerFactory.getLogger(SendServiceImpl.class);

	private IEmailService iEmailService;
	
	private MailSenderInfo mailInfo;
	
	@Resource(name="iRuleService")
	private IRuleService iRuleService;
	
	@Resource(name="kafkaSender")
	private IKafkaSender<String,String> kafkaSender;
	
	ObjectMapper mapper = new ObjectMapper();
	
	private String [] cCs = null;

	@Override
	public boolean sendObject(AbstractMonitoringInfo abstractMonitoringInfo) {
		boolean result = true;
		setCcs();
		mailInfo = new MailSenderInfo();
		mailInfo.setSubject(abstractMonitoringInfo.getSubject());
		mailInfo.setContent(abstractMonitoringInfo.getContent());
		mailInfo.setCcs(cCs);
		if (StringUtils.isNotBlank(mailInfo.getSubject()) || StringUtils.isNotBlank(mailInfo.getContent())) {
//			kafkaSender.sendMessage("email_log", 0 , DateTime.now().toDate().getTime(), SendServiceImpl.class.getName() , mailInfo, successCallback, failureCallback);
//			this.sendMessage("email_log", 0, DateTime.now().toDate().getTime(), "", mailInfo);
			logger.debug("发送成功! 数据为 : {} " , mailInfo);
		}
		return result;
	}
	
	@Override
	public boolean sendObject(AbstractMonitoringInfo abstractMonitoringInfo, int partition , String key) {
		String value = "";
		setCcs();
		mailInfo = new MailSenderInfo();
		mailInfo.setSubject(abstractMonitoringInfo.getSubject());
		mailInfo.setContent(abstractMonitoringInfo.getContent());
		mailInfo.setCcs(cCs);
		if (StringUtils.isNotBlank(mailInfo.getSubject()) || StringUtils.isNotBlank(mailInfo.getContent())) {
			try {
				value = mapper.writeValueAsString(mailInfo);
			} catch (JsonProcessingException e) {
				logger.error("" , e);
			}
			kafkaSender.sendMessage("email_log", partition , DateTime.now().toDate().getTime(), key , value, successCallback, failureCallback);
		}
		return true;
	}
	
	SuccessCallback<SendResult<String, String>> successCallback = new SuccessCallback<SendResult<String, String>>() {
		@Override
		public void onSuccess(SendResult<String, String> result) {
			logger.info("发送成功! topic : {} , ----partition :  {} " , result.getRecordMetadata().topic() , result.getRecordMetadata().partition());
		}
	};
	FailureCallback failureCallback = new FailureCallback() {
		@Override
		public void onFailure(Throwable ex) {
			logger.error("发生了异常 : " , ex);
            //失败业务逻辑
            throw new RuntimeException(ex);
		}
	};
	
	/**
	* @Title: setCcs  
	* @Description: 设置抄送邮箱
	* @return boolean    返回类型  
	 */
	private boolean setCcs() {
		boolean result = false;
		try {
			cCs = iRuleService.getCcs(new RuleBean(SystemConstants.ProductConstants.MONITORING, 0));
			if (null != cCs && cCs.length > 0) {
				result = true;
			}
		} catch (Exception e) {
			logger.error("" , e);
		}
		return result;
	}

	public IEmailService getiEmailService() {
		return iEmailService;
	}

	public void setiEmailService(IEmailService iEmailService) {
		this.iEmailService = iEmailService;
	}

}